// 版权所有2016 etcd作者
// 
// 根据Apache许可证2.0版（以下简称“许可证”）获得许可；
// 除非遵守许可证，否则不得使用此文件。
// 您可以通过
// 
// www.apache.org/licenses/LICENSE-2.0 
// 
// 除非适用法律要求或书面同意，否则根据许可证分发的软件
// 按“原样”分发，
// 无任何明示或暗示的保证或条件。
// 请参阅许可证，了解管理许可的特定语言和
// 许可证下的限制。

package grpcproxy

import (
	"context"
	"sync"
	"time"

	pb "go.etcd.io/etcd/api/v3/etcdserverpb"
	clientv3 "go.etcd.io/etcd/client/v3"

	"go.uber.org/zap"
)

// watchBroadcast将服务器监视程序广播给许多客户端监视程序。
type watchBroadcast struct {
	// 取消停止底层etcd服务器观察程序并关闭通道。
	cancel context.CancelFunc
	donec  chan struct{}

	// mu保护rev和接收器。
	mu sync.RWMutex
	// nextrev是ch上watcher的最低预期下一版本。
	nextrev int64
	// 接收器包含所有要服务的客户端watcher。
	receivers map[*watcher]struct{}
	// responses统计回复数量
	responses int
	lg        *zap.Logger
}

func newWatchBroadcast(lg *zap.Logger, wp *watchProxy, w *watcher, update func(*watchBroadcast)) *watchBroadcast {
	cctx, cancel := context.WithCancel(wp.ctx)
	wb := &watchBroadcast{
		cancel:    cancel,
		nextrev:   w.nextrev,
		receivers: make(map[*watcher]struct{}),
		donec:     make(chan struct{}),
		lg:        lg,
	}
	wb.add(w)
	go func() {
		defer close(wb.donec)

		opts := []clientv3.OpOption{
			clientv3.WithRange(w.wr.end),
			clientv3.WithProgressNotify(),
			clientv3.WithRev(wb.nextrev),
			clientv3.WithPrevKV(),
			clientv3.WithCreatedNotify(),
		}

		cctx = withClientAuthToken(cctx, w.wps.stream.Context())

		wch := wp.cw.Watch(cctx, w.wr.key, opts...)
		wp.lg.Debug("watch", zap.String("key", w.wr.key))

		for wr := range wch {
			wb.bcast(wr)
			update(wb)
		}
	}()
	return wb
}

func (wb *watchBroadcast) bcast(wr clientv3.WatchResponse) {
	wb.mu.Lock()
	defer wb.mu.Unlock()
	// 观察者从给定版本开始（如果有）；在创建
	if wb.responses > 0 || wb.nextrev == 0 {
		wb.nextrev = wr.Header.Revision + 1
	}
	wb.responses++
	for r := range wb.receivers {
		r.send(wr)
	}
	if len(wb.receivers) > 0 {
		eventsCoalescing.Add(float64(len(wb.receivers) - 1))
	}
}

// 添加时忽略标头版本如果其版本至少
// 符合广播版本，则会使观察者接收广播。如果添加，则返回true。
func (wb *watchBroadcast) add(w *watcher) bool {
	wb.mu.Lock()
	defer wb.mu.Unlock()
	if wb.nextrev > w.nextrev || (wb.nextrev == 0 && w.nextrev != 0) {
		// wb太超前了，w将错过事件
		// 或wb正在与当前观察者
		return false
	}
	if wb.responses == 0 {
		// 新创建的；创建事件将由etcd发送。
		wb.receivers[w] = struct{}{}
		return true
	}
	// 已由etcd发送；模拟创建事件
	ok := w.post(&pb.WatchResponse{
		Header: &pb.ResponseHeader{
			// todo:fill in ClusterId 
			// todo:fill in MemberId:
			Revision: w.nextrev,
			// todo:fill in RaftTerm:
		},
		WatchId: w.id,
		Created: true,
	})
	if !ok {
		return false
	}
	wb.receivers[w] = struct{}{}
	watchersCoalescing.Inc()

	return true
}
func (wb *watchBroadcast) delete(w *watcher) {
	wb.mu.Lock()
	defer wb.mu.Unlock()
	if _, ok := wb.receivers[w]; !ok {
		panic("deleting missing watcher from broadcast")
	}
	delete(wb.receivers, w)
	if len(wb.receivers) > 0 {
		// /不要将合并的唯一左侧观察者定义为dec。
		watchersCoalescing.Dec()
	}
}

func (wb *watchBroadcast) size() int {
	wb.mu.RLock()
	defer wb.mu.RUnlock()
	return len(wb.receivers)
}

func (wb *watchBroadcast) empty() bool { return wb.size() == 0 }

func (wb *watchBroadcast) stop() {
	if !wb.empty() {
		// 不要让自己成为唯一的左派观察者。
		watchersCoalescing.Sub(float64(wb.size() - 1))
	}

	wb.cancel()

	select {
	case <-wb.donec:
		// 如果客户端未能取消etcd监视程序，watchProxyStream将始终保持watchRanges全局互斥锁。
		// 这将导致监视代理无法工作。
		// 请参见pr https:
	case <-time.After(time.Second):
		wb.lg.Error("failed to cancel etcd watcher")
	}
}
